package com.example.transform;


import com.example.beans.SensorReading;
import com.example.source.SensorSource;
import com.example.source.SourceTest4_UDF;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransformTest6_Partition {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // 从文件读取数据
//        DataStream<String> inputStream = env.readTextFile("E:\\hjx\\ideapro\\myflinkstu\\src\\main\\resources\\sensor.txt");

        // 转换成SensorReading类型
//        DataStream<SensorReading> dataStream = inputStream.map(line -> {
//            String[] fields = line.split(",");
//            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
//        });

        DataStreamSource<SensorReading> inputStream = env.addSource(new SensorSource());

        // 1. shuffle
        DataStream<SensorReading> shuffleStream = inputStream.shuffle();

        shuffleStream.print("shuffle");

        // 2. keyBy
        inputStream.keyBy(line -> line.getId()).print("keyBy");

//        dataStream.keyBy("id").print("keyBy");

        // 3. global
        /*
        通过调用.global()方法，会将所
        有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行
        度变成了 1，
         */
        inputStream.global().print("global");

        // 4. 轮询分区 Round-Robin
        inputStream.rebalance().print("rebalance").setParallelism(5);

        env.execute();
    }
}
